package com.nami.gate.router.middleware.mq.consumer.factory;

import com.nami.gate.router.middleware.mq.consumer.model.ConsumerModel;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;

public class RocketMqConsumerFactory {

    private static DefaultMQPushConsumer newConsumer(final String namesrvAddr,
                                                     String consumerGroup, String topic, String tag, MessageModel messageModel) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setMessageModel(messageModel);
        consumer.subscribe(topic, tag);
        return consumer;
    }

    public static DefaultMQPushConsumer startConsumer(ConsumerModel wrapper) throws MQClientException {
        DefaultMQPushConsumer consumer = null;
        try {
            consumer = newConsumer(wrapper.getNamesrvAddr(), wrapper.getConsumerGroup(),
                wrapper.getTopic(), wrapper.getTags(), wrapper.getMessageModel());
            consumer.registerMessageListener(wrapper.getMessageListener());
            consumer.start();
            return consumer;
        } catch (Exception ex) {
            ex.printStackTrace();
            throw ex;
        }

    }


}
